package cn.doitedu.api;

import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class _16_RestartStrategy_Demo {

    public static void main(String[] args) throws Exception {

        Configuration configuration = new Configuration();
        // 指定savepoint目录，会让系统在启动时就去加载快照来恢复状态
        configuration.setString("execution.savepoint.path","file:///D:/ckpt/306f4816b3782db939e2a1298f2f37f5/chk-28");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);



        // 设置重启策略为固定延时重启
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(4, Time.seconds(2)));
        //env.setRestartStrategy(RestartStrategies.noRestart());  // 默认是不重启
        env.setRestartStrategy(RestartStrategies.fallBackRestart());

        // 开启checkpoint
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");


        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        // a b c
        stream.keyBy(s->"ok")
                .process(new KeyedProcessFunction<String, String, String>() {

                    ValueState<String> state;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        state = getRuntimeContext().getState(new ValueStateDescriptor<String>("append", String.class));
                    }

                    @Override
                    public void processElement(String value, KeyedProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {

                        if(value.equals("x")  && RandomUtils.nextInt(1,10) % 4 ==0){
                            throw new RuntimeException("哈哈哈 -----------------");
                        }

                        String old = state.value() == null? "": state.value();
                        String newStr = old + value;
                        state.update(newStr);

                        out.collect(newStr);

                    }
                }).print();

        env.execute();

    }


}
